#include <sys/socket.h>
#include "sock_connection.h"

#include <boost/bind.hpp>
#include <boost/typeof/typeof.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/thread.hpp>
#include <boost/make_shared.hpp>

#include "base/logging.h"
#include "base/url_encode.h"
#include "base/json_escape.h"
#include "base/json_parser.h"
#include "uid_sock_conn_map.h"

#include "im/adapters/message_adapter.h"
#include "im/adapters/push_adapter.h"

using namespace std;

namespace imserver {

std::string IM_REQ_END = "\r\n";
std::string IM_RES_END = "\r\n";

SockConnection::SockConnection(boost::asio::io_service& io_service)
  : io_service_(io_service)
  , socket_(io_service)
  , uid_(0)
  , version_("")
  , ready_(false)
  , valid_(false)
  , kicked_(false)
  , read_buf_begin_(0)
  , read_buf_end_(0)
  , write_buf_(NULL)
  , write_buf_begin_(0)
  , write_buf_end_(0)
{
}

SockConnection::~SockConnection() {
}

SockConnectionPtr SockConnection::Create(boost::asio::io_service& io_service) {
  return SockConnectionPtr(new SockConnection(io_service));
}

bool SockConnection::isReadCompleted(const char * data, size_t len) {
  if (!data) {
    return false;
  }
  return std::string::npos != std::string(data, len).find(IM_REQ_END);
}

bool SockConnection::parseKeyValue(const std::string& data, std::map<std::string, std::string>& output) {
  if (data.empty())
    return false;
  return base::JsonParser::instance().parse(std::string(data), output);
}

void SockConnection::Start() {
  LOG_INFO("SockConnection::Start --> begin");
  ip::tcp::no_delay nodelay(true);
  socket_.set_option(nodelay);

  boost::asio::socket_base::linger linger(true, 0);
  socket_.set_option(linger);

  boost::asio::socket_base::keep_alive keep_alive(true);
  socket_.set_option(keep_alive);

  int fd = socket_.native_handle();
  int optval = 5;
  setsockopt(fd, SOL_TCP, TCP_KEEPCNT, &optval, sizeof(optval));
  optval = 20;
  setsockopt(fd, SOL_TCP, TCP_KEEPIDLE, &optval, sizeof(optval));
  optval = 5;
  setsockopt(fd, SOL_TCP, TCP_KEEPINTVL, &optval, sizeof(optval));

  boost::asio::socket_base::send_buffer_size send_buf_size(1024);
  socket_.set_option(send_buf_size);

  valid_ = true;
  AsyncRead();
}

std::string SockConnection::remote_addr() const {
  boost::system::error_code ec;
  ip::tcp::endpoint endpoint = socket_.remote_endpoint(ec);
  if (ec) {
    // An error occurred.
    return "bad_addr";
  }
  return endpoint.address().to_string();
}


void SockConnection::Abort() {
  closeSocket();
  return;
}

void SockConnection::AsyncRead() {
  LOG_INFO("SockConnection::AsyncRead --> begin");
  socket_.async_read_some(boost::asio::buffer(read_buf_ + read_buf_end_, kReadBufLength - read_buf_end_),
      boost::bind(&SockConnection::HandleRead, shared_from_this(),
    boost::asio::placeholders::error, // 占位符
    boost::asio::placeholders::bytes_transferred));
}

void SockConnection::AsyncWrite() {
  LOG_INFO("SockConnection::AsyncWrite --> begin");
  if (socket_.is_open()) {
    boost::asio::async_write(socket_,
        boost::asio::buffer(write_buf_ + write_buf_begin_, write_buf_end_ - write_buf_begin_),
        boost::bind(&SockConnection::HandleWrite, shared_from_this(),
            boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
  } else {
    LOG_ERROR("SockConnection::AsyncWrite --> try to write on bad socket. uid:" << uid_);
    valid_ = false;
    ready_ = false;
    UidSockConnMap::instance().remove(uid_);
    //该不该去发通知呢？
  }
}

void SockConnection::WriteKickMsgAndClose(const char* data, size_t len) {
  LOG_INFO("SockConnection::WriteKickMsgAndClose --> begin");
  if (socket_.is_open()) {
    socket_.write_some(boost::asio::buffer(data, len));
  } else {
    LOG_ERROR("SockConnection::WriteKickMsgAndClose --> try to write on bad socket. uid:" << uid_);
  }
  kicked_ = true;
  closeSocket();
}

void SockConnection::HandleWrite(const boost::system::error_code& error,
    size_t bytes_transferred) {
  if (error) {
    LOG_ERROR("SockConnection::HandleWrite --> error:" << error << " errorMsg:" << error.message() << " uid:" << uid_);
    closeSocket();
    UidSockConnMap::instance().remove(uid_);
    return;
  }

  write_buf_begin_ += bytes_transferred;
  if (write_buf_begin_ < write_buf_end_) {
    AsyncWrite();
  } else {
    LOG_INFO("SockConnection::HandleWrite --> write over. content:" << std::string(write_buf_));
    //onWriteComplete(rep_str_);
  }
}

void SockConnection::HandleRead(const boost::system::error_code& error, size_t bytes_transferred) {
  LOG_INFO("SockConnection::HandleRead --> begin");
  if (error) {
    if (error != boost::asio::error::eof) {
      if (kicked_) {
        LOG_WARN("SockConnection::HandleRead --> read error by kicked. uid_:" << uid_);
      } else {
        LOG_ERROR("SockConnection::HandleRead --> error:" << error << " errorMsg:" << error.message() << " uid_:" << uid_);
      }
    }
    if (!kicked_) {
      closeSocket();
      UidSockConnMap::instance().remove(uid_);
    }
    return;
  }
  read_buf_end_ += bytes_transferred;

  if (isReadCompleted(read_buf_, read_buf_end_)) {
    if (!onReadComplete()) {
      return;
    }
  }

  AsyncRead();
}

bool SockConnection::onReadComplete() {
  std::map<std::string, std::string> read_data;
  LOG_DEBUG("SockConnection::onReadComplete --> receive:" << std::string(read_buf_, read_buf_end_));
  std::string req_str = std::string(read_buf_, read_buf_end_);
  req_str = req_str.substr(0, req_str.find(IM_REQ_END));
  reset_read_buf();
  if (!parseKeyValue(req_str, read_data)) {
    closeSocket();
    UidSockConnMap::instance().remove(uid_);
    return false;
  }
  if (!ready_) {
    if (read_data.find("uid") != read_data.end() && !read_data["uid"].empty() &&
        read_data.find("t") != read_data.end() && !read_data["t"].empty() &&
        read_data.find("v") != read_data.end() && !read_data["v"].empty() &&
        read_data.find("os") != read_data.end() && !read_data["os"].empty()) {
      uid_ = boost::lexical_cast<int64_t>(read_data["uid"]);
      SockConnectionPtr conn;
      if (UidSockConnMap::instance().get(uid_, conn)) {
        LOG_WARN("SockConnection::onReadComplete --> another login, kick old conn. receive_data:" << req_str);
        std::string error = "{\"type\":" + boost::lexical_cast<std::string>(ImMessageType::IM_MESSAGE_LOGIN) +
                            ",\"status\":2" +
                            ",\"body\":\"" + g_ImServiceCommon_constants.IM_MESSAGE_KICK_BODY + "\"}" + IM_RES_END;
        conn->WriteKickMsgAndClose(error.c_str(), error.size());
      }
      // 从read_data中取出ticket和uid，然后去验票
      if (true) {
        ready_ = true;
        UidSockConnMap::instance().add(uid_, shared_from_this());
      }
    }
    if (ready_) {
      LOG_INFO("SockConnection::onReadComplete --> verify user ok. uid:" << uid_ << " v:" << read_data["v"]
          << " os:" << read_data["os"]);
      std::string ok = "{\"type\":" + boost::lexical_cast<std::string>(ImMessageType::IM_MESSAGE_LOGIN) +
                       ",\"status\":0" +
                       ",\"body\":\"" + g_ImServiceCommon_constants.IM_MESSAGE_LOGIN_OK_BODY + "\"}" + IM_RES_END;
      if (socket_.is_open()) {
        socket_.write_some(boost::asio::buffer(ok.c_str(), ok.size()));
      }
    } else {
      LOG_WARN("SockConnection::onReadComplete --> illegal user. receive_data:" << req_str);
      std::string error = "{\"type\":" + boost::lexical_cast<std::string>(ImMessageType::IM_MESSAGE_LOGIN) +
                          ",\"status\":1" +
                          ",\"body\":\"" + g_ImServiceCommon_constants.IM_MESSAGE_LOGIN_FAILED_BODY + "\"}" + IM_RES_END;
      WriteKickMsgAndClose(error.c_str(), error.size());
      UidSockConnMap::instance().remove(uid_);
      return false;
    }
  } else {
    if (read_data.find("mid") != read_data.end() && !read_data["mid"].empty() &&
        read_data.find("did") != read_data.end() && !read_data["did"].empty() &&
        read_data.find("status") != read_data.end() && !read_data["status"].empty()) {
      int64_t msg_id = 0;
      int64_t dialogue_id = 0;
      int status = 0;
      try {
        msg_id = boost::lexical_cast<int64_t>(read_data["mid"]);
        dialogue_id = boost::lexical_cast<int64_t>(read_data["did"]);
        status = boost::lexical_cast<int>(read_data["status"]);
      } catch (std::exception& e) {
        LOG_ERROR("SockConnection::onReadComplete --> error:" << e.what() << " mid:" << read_data["mid"] <<
            " did:" << read_data["did"] << " status:" << read_data["status"]);
        status = 0;
      }
      if (status == ImMessageFlag::IM_MESSAGE_RECEIVED) {
        MessageAdapter::instance().SetMessageReceiveStatus(uid_, msg_id, dialogue_id, 2);
      } else if (status == ImMessageFlag::IM_MESSAGE_READ) {
        MessageAdapter::instance().SetMessageReadStatus(uid_, msg_id, dialogue_id, 2);
      }
    } else {
      LOG_WARN("SockConnection::onReadComplete --> illegal data. receive_data:" << req_str);
      /*
      std::string error = "{\"type\":" + boost::lexical_cast<std::string>(ImMessageType::IM_MESSAGE_READ_ERROR_DATA) +
          ",\"body\":\"" + g_ImServiceCommon_constants.IM_MESSAGE_READ_ERROR_DATA_BODY + "\"}" + IM_RES_END;
      WriteKickMsgAndClose(error.c_str(), error.size());
      UidSockConnMap::instance().remove(uid_);
      return;
      */
    }
  }
  return true;
}

}

